package com.lsh.streamgd.excelExtractor;

import com.hp.hpl.jena.rdf.model.Literal;
import com.hp.hpl.jena.rdf.model.Model;
import com.hp.hpl.jena.rdf.model.ModelFactory;
import com.hp.hpl.jena.rdf.model.Resource;
import com.hp.hpl.jena.vocabulary.RDF;
import com.lsh.streamgd.sparql.Upload;
import com.lsh.streamgd.utils.Settings;
import com.lsh.streamgd.utils.StringUtil;
import org.apache.any23.extractor.ExtractionException;
import org.apache.any23.rdf.RDFUtils;
import org.openrdf.model.URI;

import java.io.*;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel.MapMode;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicLong;

public class BigFileReader {
	private int threadSize;
	private String charset;
    private URI[] headerURIs;
    private String instanceURI, classURI;
	private int bufferSize;
    private int blockSize; //份文件存储大小
	private IHandle handle;
    private String directory;
	private ExecutorService  executorService;
	private long fileLength;
	private RandomAccessFile rAccessFile;
	private Set<StartEndPair> startEndPairs;
	private CyclicBarrier cyclicBarrier;
	private AtomicLong counter = new AtomicLong(0);

    private BigFileReader(File file,IHandle handle,String charset,int bufferSize,int threadSize, URI[] headerURIs, String instanceURI, String directory, int blockSize, String classURI){
		this.fileLength = file.length();
        this.headerURIs = headerURIs;
        this.blockSize = blockSize;
        this.instanceURI = instanceURI;
        this.classURI = classURI;
        this.directory = directory;
		this.handle = handle;
		this.charset = charset;
		this.bufferSize = bufferSize;
		this.threadSize = threadSize;
        try {
			this.rAccessFile = new RandomAccessFile(file,"r");
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		}
		this.executorService = Executors.newFixedThreadPool(threadSize);
		startEndPairs = new HashSet<StartEndPair>();
	}

    public void start(int start_index, final Map<String, Boolean> status) {
        long everySize = this.fileLength / this.threadSize;
        try {
            calculateStartEnd(start_index, everySize);
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }
        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
        final String start = df.format(new Date());// new Date()为获取当前系统时间

        final long startTime = System.currentTimeMillis();
        cyclicBarrier = new CyclicBarrier(startEndPairs.size(), new Runnable() {
            String status_content = "";

            @Override
            public void run() {
                SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
                String end = df.format(new Date());// new Date()为获取当前系统时间;
                System.out.println("use time: " + (System.currentTimeMillis() - startTime));
                System.out.println("all line: " + counter.get());
                shutdown();

                //写入数据集状态
                String path = directory + "/status.txt";
                try {
                    status_content += "Finished" + "\n";
                    status_content += "Start Time: " + start + "\n";
                    status_content += "End Time: " + end + "\n";
                    status_content += "Use Time: " + (System.currentTimeMillis() - startTime) + "ms \n";
                    status_content += "All Line : " + counter.get();
                    FileOutputStream out = new FileOutputStream(path);
                    out.write(status_content.getBytes());
                } catch (FileNotFoundException e) {
                    e.printStackTrace();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                status.put("finished", true);
            }
        });

        for (StartEndPair pair : startEndPairs) {
            System.out.println("分配分片：" + pair);
            this.executorService.execute(new SliceReaderTask(pair, headerURIs, directory));
        }
    }

    private void calculateStartEnd(long start,long size) throws IOException{
		if(start>fileLength-1){
			return;
		}
		StartEndPair pair = new StartEndPair();
		pair.start=start;
		long endPosition = start+size-1;
		if(endPosition>=fileLength-1){
			pair.end=fileLength-1;
			startEndPairs.add(pair);
			return;
		}
		
		rAccessFile.seek(endPosition);
		byte tmp =(byte) rAccessFile.read();
		while(tmp!='\n' && tmp!='\r'){
			endPosition++;
			if(endPosition>=fileLength-1){
				endPosition=fileLength-1;
				break;
			}
			rAccessFile.seek(endPosition);
			tmp =(byte) rAccessFile.read();
		}
		pair.end=endPosition;
		startEndPairs.add(pair);
		
		calculateStartEnd(endPosition+1, size);
	}

	public void shutdown(){
		try {
			this.rAccessFile.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		this.executorService.shutdown();
	}

	private String handle(byte[] bytes) throws IOException, ExtractionException {
		String line = null;
		if(this.charset==null){
			line = new String(bytes);
		}else{
			line = new String(bytes,charset);
		}
		if(line!=null && !"".equals(line)){
			this.handle.handle(line);
			counter.incrementAndGet();
		}

        return line;
	}

	private static class StartEndPair{
		public long start;
		public long end;
		
		@Override
		public String toString() {
			return "start="+start+";end="+end;
		}

		@Override
		public int hashCode() {
			final int prime = 31;
			int result = 1;
			result = prime * result + (int) (end ^ (end >>> 32));
			result = prime * result + (int) (start ^ (start >>> 32));
			return result;
		}

		@Override
		public boolean equals(Object obj) {
			if (this == obj)
				return true;
			if (obj == null)
				return false;
			if (getClass() != obj.getClass())
				return false;
			StartEndPair other = (StartEndPair) obj;
			if (end != other.end)
				return false;
			if (start != other.start)
				return false;
			return true;
		}
	}

	private class SliceReaderTask implements Runnable{
		private long start;
		private long sliceSize;
        private URI[] headerURIs;
        private String directory;
		private byte[] readBuff;

		public SliceReaderTask(StartEndPair pair, URI[] headerURIs, String directory) {
			this.start = pair.start;
            this.headerURIs = headerURIs;
            this.directory = directory;
			this.sliceSize = pair.end-pair.start+1;
			this.readBuff = new byte[bufferSize];
		}

        private void produceRowStatements(
                URI rowSubject,
                String[] values,
                Model model
        ) {
            int index = 0;
            for (String cell : values) {
                if (cell.startsWith("\"") && cell.endsWith("\"")) cell = cell.substring(1, cell.length() - 1);

                if (index >= headerURIs.length) {
                    // there are some row cells that don't have an associated column name
                    break;
                }
                if (cell.equals("")) {
                    index++;
                    continue;
                }
                URI predicate = headerURIs[index];

                String lang = "", p="";

                if (predicate.getLocalName().contains("@")) {
                    p = predicate.toString().split("@")[0];
                    lang = predicate.toString().split("@")[1];
                } else {
                    p = predicate.toString();
                }

                Resource res = model.createResource(rowSubject.toString());
                if (RDFUtils.isAbsoluteURI(cell)) {
                    model.add(res, model.createProperty(p), model.createResource(cell));
                } else {
                    Literal object = getObjectFromCell(model, cell, lang);

                    if (StringUtil.isBlank(lang)) {
                        model.add(res, model.createProperty(p), object);
                    } else {
                        model.add(res, model.createProperty(p), object);
                    }
                }

                model.add(res, RDF.type, model.createResource(classURI));

                index++;
            }
        }

        private Literal getObjectFromCell(Model model, String cell, String lang) {
            cell = cell.trim();

            if (StringUtil.isBlank(lang)) {
                return model.createTypedLiteral(cell);
            } else {
                //多语言
                return model.createLiteral(cell, lang);
            }
        }

		@Override
		public void run() {
            Model model = ModelFactory.createDefaultModel();
			try {
                String name = UUID.randomUUID().toString();

                Upload upload = new Upload(Settings.graphDriver, Settings.graphHost, Settings.graphUsername, Settings.graphPassword);

                MappedByteBuffer mapBuffer = rAccessFile.getChannel().map(MapMode.READ_ONLY,start, this.sliceSize);
				ByteArrayOutputStream bos = new ByteArrayOutputStream();
				for(int offset=0;offset<sliceSize;offset+=bufferSize){
					int readLength;
					if(offset+bufferSize<=sliceSize){
						readLength = bufferSize;
					}else{
						readLength = (int) (sliceSize-offset);
					}
					mapBuffer.get(readBuff, 0, readLength);

					for(int i=0;i<readLength;i++){
						byte tmp = readBuff[i];
						if(tmp=='\n' || tmp=='\r'){
                            URI rowSubject = RDFUtils.uri(
                                    instanceURI.toString(),
                                    "/" + UUID.randomUUID().toString().replaceAll("-", "")
                            );

                            String line = handle(bos.toByteArray());

                            if (!StringUtil.isBlank(line)) {
                                String[] cells = CSVUtils.parseLine(line).toArray(new String[]{});
                                produceRowStatements(rowSubject, cells, model);
                            }
							bos.reset();
						}else{
							bos.write(tmp);
						}
					}

                    if (model.size() > blockSize) {
                        String path = directory + "/" + name + ".n3";
                        Writer out = new BufferedWriter(new OutputStreamWriter(
                                new FileOutputStream(path), "UTF8"));
//                        model.write(out, "TTL"); //写入文件
                        upload.convert2RDF(model); //传入LSH
                        model.close();
                        System.out.println("Finished file [" + name + "]");
                        name = UUID.randomUUID().toString();
                        model = ModelFactory.createDefaultModel();
                    }
				}
				if(bos.size()>0){
                    URI rowSubject = RDFUtils.uri(
                            instanceURI.toString(),
                            "/" + UUID.randomUUID().toString().replaceAll("-", "")
                    );

                    String line = handle(bos.toByteArray());

                    if (!StringUtil.isBlank(line)) {
                        String[] cells = CSVUtils.parseLine(line).toArray(new String[]{});
                        produceRowStatements(rowSubject, cells, model);
                    }
                }

                if (!model.isEmpty()) {
                    name = UUID.randomUUID().toString();
                    String path = directory + "/" + name + ".n3";
                    Writer out = new BufferedWriter(new OutputStreamWriter(
                        new FileOutputStream(path), "UTF8"));
//                    model.write(out, "TTL"); //写入文件
                    upload.convert2RDF(model); //传入LSH
                    System.out.println("Finished file [" + name + "]");
                }
                model.close();

                cyclicBarrier.await();//测试性能用
			}catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

    public static class Builder{
		private int threadSize=1;
		private String charset=null;
        private URI[] headerURIs;
        private String instanceURI, classURI;
        private int blockSize;
        private String directory;
		private int bufferSize=1024*1024;
		private IHandle handle;
		private File file;
		public Builder(String file,IHandle handle){
			this.file = new File(file);
			if(!this.file.exists())
				throw new IllegalArgumentException("文件不存在！");
			this.handle = handle;
		}
		
		public Builder withThreadSize(int size){
			this.threadSize = size;
			return this;
		}

        public Builder withHeaderURIs(URI[] headerURIs){
            this.headerURIs = headerURIs;
            return this;
        }

        public Builder withCharset(String charset){
			this.charset= charset;
			return this;
		}
		
		public Builder withBufferSize(int bufferSize){
			this.bufferSize = bufferSize;
			return this;
		}

        public Builder withInstanceURI(String instanceURI){
            this.instanceURI = instanceURI;
            return this;
        }

        public Builder withClassURI(String classURI){
            this.classURI = classURI;
            return this;
        }

        public Builder withDirectory(String directory){
            this.directory = directory;
            return this;
        }

        public Builder withBlockSize(int blockSize){
            this.blockSize = blockSize;
            return this;
        }

		public BigFileReader build(){
			return new BigFileReader(this.file,this.handle,this.charset,this.bufferSize, this.threadSize, this.headerURIs, this.instanceURI, this.directory, this.blockSize, this.classURI);
		}
	}
}
